import os
from urllib.parse import quote_plus

import pandas as pd
from sqlalchemy import create_engine

import streamlit as st

os.environ["STREAMLIT_SERVER_ENABLE_STATIC_SERVING"] = "true"

@st.cache_resource
def get_engine():
    host = st.secrets["mysql"]['host']
    user = st.secrets["mysql"]['user']
    password = st.secrets["mysql"]['password']
    db = st.secrets["mysql"]['datebase_zhongtai']
    connstr = f"mysql+pymysql://{user}:%s@{host}:3306/{db}?charset=utf8" % quote_plus(f'{password}')
    # 显式配置连接池参数，根据需要调整 pool_size / max_overflow
    engine = create_engine(
        connstr,
        pool_size=20,
        max_overflow=40,
        pool_recycle=3600,
        pool_pre_ping=True,
        pool_timeout=30
    )
    return engine


def select_goods(erp_sku):
    """通过 erp_sku 查询商品基本信息"""
    if erp_sku is None or erp_sku.strip() == '':
        raise ValueError(f"erp_sku {erp_sku} 不存在.")
    sql = f"select * from item_basic where erp_sku='{erp_sku}' limit 1"
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        raise ValueError(f"erp_sku {erp_sku} 不存在.")
    return df.to_dict('records')[0]


def select_logistics_types(country):
    """通过国家查询支持的物流类型"""
    sql = f"""select warehouse, express, supplier from supplier_express_house_new where country='{country}' and flag = 1 group by warehouse, express, supplier"""
    df_logistics_types = pd.read_sql(sql, con=get_engine())
    if df_logistics_types.empty:
        raise ValueError(f"国家 {country} 无支持的物流类型.")
    json_list = df_logistics_types.to_dict('records')
    return json_list


def select_type_of_zip(country, express, supplier, zip):
    """通过快递公司、国家、省份、供应商查询邮编类型"""
    conditions = [
        "country = %(country)s",
        "str_zipcode <= %(zip)s",
        "end_zipcode >= %(zip)s"
    ]
    params = {
        "country": country,
        "zip": zip
    }
    if express is not None:
        conditions.append("express = %(express)s")
        params["express"] = express

    if supplier is not None:
        conditions.append("supplier = %(supplier)s")
        params["supplier"] = supplier

    where_clause = " and ".join(conditions)
    sql = f"select type_of_zip from zip_code_type where {where_clause} limit 1"
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]['type_of_zip']


def select_non_deliverable_zone(supplier, express, zip):
    """通过快递公司、国家、省份、供应商查询不可送达区域"""
    sql = f"""select id from non_deliverable_zone where express='{express}' and supplier='{supplier}' 
    and str_zip <= '{zip}' and end_zip >= '{zip}' limit 1"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return None
    return df.to_dict('records')


def select_ca_cost_type_surcharge(supplier, express, oversea_warehouse, zip_type):
    """通过快递公司、承运商、海外仓、货型查询加拿大邮编类型附加费"""
    sql = f"""select expenses from ca_cost where experss='{express}' and supplier='{supplier}' 
    and cost_type='{zip_type}' and warehouse='{oversea_warehouse}' limit 1"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return 0.0
    return df.to_dict('records')[0]['expenses']


def select_zone(oversea_warehouse, zip):
    """通过邮编查询区域"""
    sql = f"select zone from zip_zone_us where oversea_warehouse='{oversea_warehouse}' and zip_code = '{zip}' limit 1"
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return None
    return df.to_dict('records')[0]['zone']


def select_zone_ca(zip, supplier, express):
    """通过加拿大邮编查询区域"""
    if zip is None or zip.strip() == '':
        raise ValueError(f"zip {zip} 不存在.")
    sql = (f"select zone from zip_zone_ca where supplier='{supplier}' and express='{express}' and str_zipcode <= '{zip}' and end_zipcode >= '{zip}' limit 1")
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        raise ValueError(f"zip {zip} 不存在.")
    return df.to_dict('records')[0]['zone']


def select_us_cost_type_good_surcharge(supplier, express, zone, goods_type):
    """通过快递公司、承运商、区域、货型查询美国邮编类型附加费（参数化查询）"""
    sql = """
        SELECT warehouse, expenses, str_billed_weight, end_billed_weight
        FROM us_cost
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND cost_type = %(goods_type)s
          AND str_zone <= %(zone)s
          AND end_zone >= %(zone)s
    """
    params = {
        "supplier": supplier,
        "express": express,
        "goods_type": goods_type,
        "zone": zone
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return 0.0
    return df.to_dict('records')

def select_ca_cost_type_good_surcharge(supplier, express, orversea_warehouse, goods_type):
    """通过快递公司、承运商、海外仓、货型查询加拿大邮编类型附加费"""
    sql = f"""select expenses from ca_cost where supplier='{supplier}' and experss='{express}' 
        and cost_type='{goods_type}'  and warehouse = '{orversea_warehouse}';"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return 0.0
    return df.to_dict('records')


def select_eu_cost_type_good_surcharge(supplier, express, goods_type):
    """通过快递公司、承运商、货型查询欧盟邮编类型附加费"""
    sql = f"""select expenses from eu_cost where supplier='{supplier}' and express='{express}' 
        and cost_type='{goods_type}';"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return 0.0
    return df.to_dict('records')


def select_special_zip_surcharge_type(country, express, state, supplier, zip):
    """查询特殊邮编附加费类型"""
    conditions = [
        "country = %(country)s",
        "str_zipcode <= %(zip)s",
        "end_zipcode >= %(zip)s"
    ]
    params = {
        "country": country,
        "zip": zip
    }
    if express:
        conditions.append("express = %(express)s")
        params["express"] = express
    if state:
        conditions.append("prefecture = %(state)s")
        params["state"] = state
    if supplier:
        conditions.append("supplier = %(supplier)s")
        params["supplier"] = supplier

    where_clause = " and ".join(conditions)
    sql = f"select type_of_zip from zip_code_type where {where_clause};"
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return 0.0
    return df.to_dict('records')[0]['type_of_zip']


def select_non_deliverable_zone(supplier, express, zip):
    sql = """
          SELECT id
          FROM non_deliverable_zone
          WHERE str_zip IS NOT NULL
            AND supplier = %(supplier)s
            AND express = %(express)s
            AND str_zip = %(zip)s
            AND end_zip >= %(zip)s 
          """
    params = {
        "supplier": supplier,
        "express": express,
        "zip": zip
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return 0.0
    return df.to_dict('records')


def select_us_cost_type_surcharge(supplier, express, zip_type):
    """通过快递公司、承运商、区域、货型查询欧洲邮编类型附加费"""
    sql = f"""SELECT warehouse, str_billed_weight, end_billed_weight, expenses FROM `us_cost` WHERE supplier = '{supplier}' 
    AND express = '{express}' AND cost_type = '{zip_type}'"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return None
    return df.to_dict('records')


def select_eu_cost_type_surcharge(supplier, express, goods_type):
    """通过快递公司、承运商、区域、货型查询美国邮编类型附加费"""
    sql = f"""SELECT country, expenses, additional_expenses FROM eu_cost WHERE 
    supplier='{supplier}' and express='{express}' and cost_type='{goods_type}';"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return 0.0
    return df.to_dict('records')


def select_eu_country_name(country):
    """通过国家简码查询国家全称"""
    sql = f"""select country from eu_country where country_code = '{country}' and country_code != 'DE' and flag = 1"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return None
    return df.to_dict('records')[0]['country']


def select_peck_season_surcharge(supplier, express, goods_type=None):
    """
    通过快递公司、承运商、（可选）货型查询旺季附加费
    """
    sql = """
        SELECT cost_type, expenses, goods_type as good_type, rate
        FROM peak_season_surcharge
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND str_date <= NOW()
          AND end_date >= NOW()
    """
    params = {"supplier": supplier, "express": express}

    if goods_type is not None:
        sql += " AND (goods_type = %(goods_type)s OR goods_type IS NULL)"
        params["goods_type"] = goods_type
    else:
        sql += " AND goods_type IS NULL"

    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')


def select_discount(supplier, express):
    """通过快递公司、承运商查询折扣"""
    sql = f"""SELECT discount FROM discount_charge WHERE supplier = '{supplier}' AND express = '{express}' and flag = 1"""
    df = pd.read_sql(sql, con=get_engine())
    if df.empty:
        return None
    return df.to_dict('records')[0]['discount']


def select_ctt_zone1(zip_code):
    """通过邮编查询区域"""
    if zip_code is None or str(zip_code).strip() == "":
        return None
    sql = """
        SELECT zone, islands
        FROM ctt_zone
        WHERE str_zip <= %(zip)s
          AND end_zip >= %(zip)s
        LIMIT 1
    """
    params = {"zip": str(zip_code)}
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]


def select_eu_express_expenses(supplier, express, country, billing_weight, region=None, islands=None):
    conditions = [
        "supplier = %(supplier)s",
        "express = %(express)s",
        "country = %(country)s",
        "str_billed_weight <= %(billing_weight)s",
        "end_billed_weight >= %(billing_weight)s"
    ]
    params = {
        "supplier": supplier,
        "express": express,
        "country": country,
        "billing_weight": billing_weight
    }
    if region is not None:
        conditions.append("region = %(region)s")
        params["region"] = region
    if islands is not None:
        conditions.append("islands = %(islands)s")
        params["islands"] = islands

    where_clause = " and ".join(conditions)
    sql = f"""
        SELECT basic_expenses AS expenses,
               additional_weight_expenses AS additional_weight,
               end_billed_weight AS end_weight
        FROM eu_express_expenses
        WHERE {where_clause}
        ORDER BY id
        LIMIT 1
    """
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]


def select_us_delivery_cost(supplier, express, zone, billing_weight):
    sql = f"""
        SELECT warehouse, expenses
        FROM us_express_expenses
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND zone = %(zone)s
          AND billed_weight = CEIL(%(billing_weight)s)
    """
    params = {
        "supplier": supplier,
        "express": express,
        "zone": zone,
        "billing_weight": billing_weight
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')


def select_ca_delivery_cost(supplier, express, zone, billing_weight):
    sql = f"""
        SELECT warehouse, expenses
        FROM ca_express_expenses
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND zone = %(zone)s
          AND billed_weight = CEIL(%(billing_weight)s)
    """
    params = {
        "supplier": supplier,
        "express": express,
        "zone": zone,
        "billing_weight": billing_weight
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')


def select_region(country, zip):
    sql = """
        SELECT region
        FROM region_zip
        WHERE country = %(country)s
          AND str_zip <= %(zip)s
          AND end_zip >= %(zip)s
    """
    params = {
        "country": country,
        "zip": zip
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]['region']


def select_eu_delivery_cost(supplier, express, country, billing_weight, region=None, islands=None):
    conditions = [
        "supplier = %(supplier)s",
        "express = %(express)s",
        "country = %(country)s",
        "str_billed_weight <= %(billing_weight)s",
        "end_billed_weight >= %(billing_weight)s"
    ]
    params = {
        "supplier": supplier,
        "express": express,
        "country": country,
        "billing_weight": billing_weight
    }
    if region is not None:
        conditions.append("region = %(region)s")
        params["region"] = region
    if islands is not None:
        conditions.append("islands = %(islands)s")
        params["islands"] = islands

    where_clause = " and ".join(conditions)
    sql = f"""
        SELECT basic_expenses AS expenses,
               additional_weight_expenses AS additional_weight,
               end_billed_weight AS end_weight
        FROM eu_express_expenses
        WHERE {where_clause}
        ORDER BY id
        LIMIT 1
    """
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]


def select_rush_gls_zip(country, zip):
    sql = """
        SELECT type
        FROM rush_gls_zip
        WHERE country = %(country)s
          AND str_zip <= %(zip)s
          AND end_zip >= %(zip)s
    """
    params = {
        "country": country,
        "zip": zip
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]['type']


def select_house_cost(supplier, express, country, billing_weight):
    """
    查询 house_cost 表的费用
    """
    sql = """
        SELECT expenses
        FROM house_cost
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND country = %(country)s
          AND str_billed_weight <= %(billing_weight)s
          AND end_billed_weight >= %(billing_weight)s
        ORDER BY id
        LIMIT 1
    """
    params = {
        "supplier": supplier,
        "express": express,
        "country": country,
        "billing_weight": billing_weight
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]['expenses']


def select_energy_additional_cost(supplier, express):
    """
    查询 energy_additional 表的燃油附加费 rate
    """
    sql = """
        SELECT rate
        FROM energy_additional
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND str_time <= NOW()
          AND end_time >= NOW()
          AND flag = 1
    """
    params = {
        "supplier": supplier,
        "express": express
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]['rate']


def select_bunker_surcharge(country, express=None, supplier=None):
    """
    查询 bunker_surcharge 表的 fsc（燃油附加费）
    """
    sql = """
        SELECT fsc
        FROM bunker_surcharge
        WHERE region = %(country)s
          AND flag = 1
          AND fsc != 0 \
        """
    select_country = country if country in ['US', 'CA'] else 'DE'
    params = {"country": select_country}
    if express is not None:
        sql += " AND express = %(express)s"
        params["express"] = express
    if supplier is not None and country != 'US':
        sql += " AND supplier = %(supplier)s"
        params["supplier"] = supplier

    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]['fsc']


def select_customs_clearance_fee(country, express, supplier):
    """
    查询 customs_clearance_fee 表的详细清关费信息
    """
    sql = """
        SELECT cost_type AS cost_type,
               country,
               str_zip,
               end_zip,
               expenses
        FROM customs_clearance_fee
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND country = %(country)s
          AND flag = 1
    """
    params = {
        "supplier": supplier,
        "express": express,
        "country": country
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')


def select_other_additional_cost(country, express, supplier):
    """
    查询 other_additional 表的费用
    """
    sql = """
        SELECT expenses, rate
        FROM other_additional
        WHERE supplier = %(supplier)s
          AND express = %(express)s
          AND country = %(country)s
          AND flag = 1
    """
    params = {
        "supplier": supplier,
        "express": express,
        "country": country
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]


def select_ca_tax(state, supplier):
    """
    查询 ca_tax 表的加拿大税 tax
    """
    sql = """
        SELECT tax
        FROM ca_tax
        WHERE state = %(state)s
          AND supplier = %(supplier)s
          AND flag = 1
    """
    params = {
        "state": state,
        "supplier": supplier
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return 0.0
    return float(df.to_dict('records')[0]['tax'])


def select_self_delivery_cost_by_weight(supplier, warehouse, billing_weight):
    """
    查询 self_delivery_cost 表的费用
    """
    sql = """
        SELECT cost
        FROM self_delivery_cost
        WHERE supplier = %(supplier)s
          AND warehouse = %(warehouse)s
          AND str_weight <= %(billing_weight)s
          AND weight > %(billing_weight)s
          AND flag = 1
        LIMIT 1
    """
    params = {
        "supplier": supplier,
        "warehouse": warehouse,
        "billing_weight": billing_weight
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return 0.0
    return df.to_dict('records')[0]['cost']


def select_self_delivery_cost(supplier, warehouse):
    """
    查询 self_delivery_cost 表的 cost
    """
    sql = """
        SELECT cost
        FROM self_delivery_cost
        WHERE supplier = %(supplier)s
          AND warehouse = %(warehouse)s
          AND flag = 1
    """
    params = {
        "supplier": supplier,
        "warehouse": warehouse
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return 0.0
    return df.to_dict('records')[0]['cost']


def select_out_bound_cost(supplier, country, gross_weight):
    """
    查询出库费用 out_warehouse_expenses
    """
    sql = """
        SELECT expenses,
               base_weight,
               additional_weight
        FROM out_warehouse_expenses
        WHERE supplier = %(supplier)s
          AND country = %(country)s
          AND str_weight <= %(gross_weight)s
          AND weight >= %(gross_weight)s
        ORDER BY id
        LIMIT 1
    """
    params = {
        "supplier": supplier,
        "country": country,
        "gross_weight": gross_weight
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')[0]


def select_delivery_service(country, warehouse, express):
    """通过国家、海外仓、快递查询可用交付服务"""
    sql = """
        SELECT goods_type,
               delivery_service,
               service_code
        FROM delivery_service
        WHERE country = %(country)s
          AND warehouse = %(warehouse)s
          AND express = %(express)s
    """
    params = {
        "country": country,
        "warehouse": warehouse,
        "express": express
    }
    df = pd.read_sql(sql, con=get_engine(), params=params)
    if df.empty:
        return None
    return df.to_dict('records')
